package com.niodata.dt.fs;

import java.io.IOException;
import java.net.URI;
import java.security.PrivilegedExceptionAction;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocalFileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.util.Progressable;

/**
 * proxy hdfs filesystem,support local filesystem and distributed filesystem.
 */
public class ProxyHdfsFileSystem extends FileSystem {

  private final UserGroupInformation ugi;
  private final String user;
  private FileSystem localFileSystem;

  public ProxyHdfsFileSystem(String user, UserGroupInformation realUser) {
    this.user = user;
    ugi = UserGroupInformation.createProxyUser(user, realUser);
  }

  public ProxyHdfsFileSystem(String user) throws IOException {
    this(user, UserGroupInformation.getCurrentUser());
  }

  public void setFileSystem(FileSystem fileSystem) {
    this.localFileSystem = fileSystem;
  }

  @Override
  public URI getUri() {
    return localFileSystem.getUri();
  }

  @Override
  public FSDataInputStream open(Path f, int bufferSize) throws IOException {
    try {
      return ugi.doAs(new PrivilegedExceptionAction<FSDataInputStream>() {
        @Override
        public FSDataInputStream run() throws Exception {
          return localFileSystem.open(f, bufferSize);
        }
      });
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    return null;
  }

  @Override
  public FSDataOutputStream create(Path f, FsPermission permission, boolean overwrite,
        int bufferSize, short replication, long blockSize, Progressable progress)
        throws IOException {
    try {
      return ugi.doAs(new PrivilegedExceptionAction<FSDataOutputStream>() {
        @Override
        public FSDataOutputStream run() throws Exception {
          FSDataOutputStream fs = localFileSystem.create(f, permission,
                overwrite,
                bufferSize,
                replication,
                blockSize,
                progress);
          fs.close();
          localFileSystem.setOwner(f, user, ugi.getPrimaryGroupName());
          return localFileSystem.create(f, permission,
                true,
                bufferSize,
                replication,
                blockSize,
                progress);
        }
      });
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    return null;
  }

  @Override
  public FSDataOutputStream append(Path f, int bufferSize, Progressable progress)
        throws IOException {
    try {
      return ugi.doAs(new PrivilegedExceptionAction<FSDataOutputStream>() {
        @Override
        public FSDataOutputStream run() throws Exception {
          return localFileSystem.append(f, bufferSize, progress);
        }
      });
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    return null;
  }

  @Override
  public boolean rename(Path src, Path dst) throws IOException {
    try {
      return ugi.doAs(new PrivilegedExceptionAction<Boolean>() {
        @Override
        public Boolean run() throws Exception {
          return localFileSystem.rename(src, dst);
        }
      });
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    return false;
  }

  @Override
  public boolean delete(Path f, boolean recursive) throws IOException {
    try {
      return ugi.doAs(new PrivilegedExceptionAction<Boolean>() {
        @Override
        public Boolean run() throws Exception {
          return localFileSystem.delete(f, recursive);
        }
      });
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    return false;
  }

  @Override
  public FileStatus[] listStatus(Path f) throws IOException {
    try {
      return ugi.doAs(new PrivilegedExceptionAction<FileStatus[]>() {
        @Override
        public FileStatus[] run() throws Exception {
          return localFileSystem.listStatus(f);
        }
      });
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    return null;
  }

  @Override
  public Path getWorkingDirectory() {
    return localFileSystem.getWorkingDirectory();
  }

  @Override
  public void setWorkingDirectory(Path nweDir) {
    localFileSystem.setWorkingDirectory(nweDir);
  }

  @Override
  public boolean mkdirs(Path f, FsPermission permission) throws IOException {
    try {
      return ugi.doAs(new PrivilegedExceptionAction<Boolean>() {
        @Override
        public Boolean run() throws Exception {
          boolean succ = localFileSystem.mkdirs(f, permission);
          if (succ && !localFileSystem.getClass().equals(LocalFileSystem.class)) {
            localFileSystem.setOwner(f, user, user);
          }
          return succ;
        }
      });
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    return false;
  }

  @Override
  public FileStatus getFileStatus(Path f) throws IOException {
    try {
      return ugi.doAs(new PrivilegedExceptionAction<FileStatus>() {
        @Override
        public FileStatus run() throws Exception {
          return localFileSystem.getFileLinkStatus(f);
        }
      });
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    return null;
  }

  @Override
  public Configuration getConf() {
    return localFileSystem.getConf();
  }
}
